package host.skiree.nettydemo.nio.c4;

import host.skiree.nettydemo.ByteBufferUtil;
import lombok.extern.slf4j.Slf4j;
import org.junit.jupiter.api.Test;
import org.springframework.boot.test.context.SpringBootTest;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.*;
import java.nio.charset.Charset;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @author gcl
 * @date 2022/9/6 10:02
 */
@Slf4j
@SpringBootTest
public class NettyDemo {

    /**
     * 用来测试
     */
    @Test
    void ServerTest() throws IOException {
        ByteBuffer buffer = ByteBuffer.allocate(16);
        // 创建服务器
        ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
        // 绑定监听端口
        serverSocketChannel.bind(new InetSocketAddress(8080));
        // 创建集合用于放置与客户端的连接
        List<SocketChannel> socketChannels = new ArrayList<>();
        while (true) {
            log.debug("connecting...");
            // 建立与客户端连接 SocketChannel用来与客户端之间通信 使用accept时线程会停止运行进入阻塞状态，等待客户端连接
            SocketChannel socketChannel = serverSocketChannel.accept();
            log.debug("connected...{}", socketChannel);
            socketChannels.add(socketChannel);
            // 接受客户端发送的数据
            for (SocketChannel channel: socketChannels) {
                // 接受客户端发送的数据到buffer缓冲区
                log.debug("before read...{}",channel);
                channel.read(buffer); // read也是阻塞方法，等待客户端发送数据
                buffer.flip(); // 切换到读模式
                ByteBufferUtil.debugAll(buffer);
                buffer.clear(); // 清空缓冲区进入到写模式
                log.debug("after read...{}",channel);
            }
        }
    }

    /**
     * 用来测试
     */
    @Test
    void ServerTestNoBlock() throws IOException {
        ByteBuffer buffer = ByteBuffer.allocate(16);
        // 创建服务器
        ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
        // 改为非阻塞模式
        serverSocketChannel.configureBlocking(false);
        // 绑定监听端口
        serverSocketChannel.bind(new InetSocketAddress(8080));
        // 创建集合用于放置与客户端的连接
        List<SocketChannel> socketChannels = new ArrayList<>();
        while (true) {
            // 建立与客户端连接 SocketChannel用来与客户端之间通信 非阻塞模式 不阻塞方法，为空返回null
            SocketChannel socketChannel = serverSocketChannel.accept();
            // 判断是否有客户端连接
            if (socketChannel != null){
                log.debug("connected...{}", socketChannel);
                socketChannels.add(socketChannel);
            }
            // 接受客户端发送的数据
            for (SocketChannel channel: socketChannels) {
                // 设置非阻塞模式
                channel.configureBlocking(false);
                // 接受客户端发送的数据到buffer缓冲区
                long len = channel.read(buffer); // 非阻塞模式 read不阻塞方法，为空返回0
                if ( len > 0 ){
                    buffer.flip(); // 切换到读模式
                    ByteBufferUtil.debugAll(buffer);
                    buffer.clear(); // 清空缓冲区进入到写模式
                    log.debug("after read...{}",channel);
                }
            }
        }
    }

    /**
     * 用来测试
     */
    @Test
    void ServerSelectorTest() throws IOException {
        // 创建Selector对象
        Selector selector = Selector.open();
        // 创建服务器
        ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
        // 改为非阻塞模式
        serverSocketChannel.configureBlocking(false);
        // 建立selector和channel的连接
        // selectionKey 就是将来事件发生后 通过他可以知道事件和那个channel相关 共有四种事件 1.accept 2.connect 3.read 4.write
        SelectionKey selectionKey =  serverSocketChannel.register(selector, 0, null);
        // selectionKey 只关注 accept事件
        selectionKey.interestOps(SelectionKey.OP_ACCEPT);
        log.debug("register key:{}", selectionKey);
        // 绑定监听端口
        serverSocketChannel.bind(new InetSocketAddress(8080));
        while (true) {
            // select方法，没有事件线程阻塞，有事件线程才会恢复运行
            selector.select();
            // 处理事件
            Iterator<SelectionKey> selectionKeys = selector.selectedKeys().iterator();
            while (selectionKeys.hasNext()){
                SelectionKey key = selectionKeys.next();
                log.debug("key:{}", key);
                if ( key.isAcceptable() ){
                    ServerSocketChannel serverSocketChannel1 = (ServerSocketChannel) key.channel();
                    SocketChannel socketChannel1 = serverSocketChannel1.accept();
                    socketChannel1.configureBlocking(false);
                    SelectionKey selectionKey1 = socketChannel1.register(selector, 0, null);
                    selectionKey1.interestOps(SelectionKey.OP_READ);
                    log.debug("channel:{}", socketChannel1);
                }else if ( key.isReadable() ){
                    try{
                        ByteBuffer buffer = ByteBuffer.allocate(16);
                        SocketChannel channel = (SocketChannel) key.channel();
                        long len = channel.read(buffer);
                        if ( len == -1 ){
                            // 客户端正常断开 服务端接收到的字节数为-1
                            key.cancel();
                        }else {
                            buffer.flip();
                            ByteBufferUtil.debugAll(buffer);
                            buffer.clear();
                        }
                    }catch (IOException e){
                        e.printStackTrace();
                        key.cancel();
                    }
                }
                // 删除已经处理完成的事件
                selectionKeys.remove();
            }
        }
    }

    /**
     * 用来测试
     */
    @Test
    void ServerSelectorMsgSize() throws IOException {
        // 创建Selector对象
        Selector selector = Selector.open();
        // 创建服务器
        ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
        // 改为非阻塞模式
        serverSocketChannel.configureBlocking(false);
        // 建立selector和channel的连接
        // selectionKey 就是将来事件发生后 通过他可以知道事件和那个channel相关 共有四种事件 1.accept 2.connect 3.read 4.write
        SelectionKey selectionKey =  serverSocketChannel.register(selector, 0, null);
        // selectionKey 只关注 accept事件
        selectionKey.interestOps(SelectionKey.OP_ACCEPT);
        log.debug("register key:{}", selectionKey);
        // 绑定监听端口
        serverSocketChannel.bind(new InetSocketAddress(8080));
        while (true) {
            // select方法，没有事件线程阻塞，有事件线程才会恢复运行
            selector.select();
            // 处理事件
            Iterator<SelectionKey> selectionKeys = selector.selectedKeys().iterator();
            while (selectionKeys.hasNext()){
                SelectionKey key = selectionKeys.next();
                log.debug("key:{}", key);
                if ( key.isAcceptable() ){
                    ServerSocketChannel serverSocketChannel1 = (ServerSocketChannel) key.channel();
                    SocketChannel socketChannel1 = serverSocketChannel1.accept();
                    socketChannel1.configureBlocking(false);
                    ByteBuffer buffer = ByteBuffer.allocate(16);
                    // 注册Channel attachment 将buffer作为附件也注册到selector中
                    SelectionKey selectionKey1 = socketChannel1.register(selector, 0, buffer);
                    selectionKey1.interestOps(SelectionKey.OP_READ);
                    log.debug("channel:{}", socketChannel1);
                }else if ( key.isReadable() ){
                    try{
                        // 从这里取 注册时传入的附件 bytebuffer 缓冲区 key.attach() 绑定新的附件
                        ByteBuffer buffer = (ByteBuffer) key.attachment();
                        SocketChannel channel = (SocketChannel) key.channel();
                        long len = channel.read(buffer);
                        if ( len == -1 ){
                            // 客户端正常断开 服务端接收到的字节数为-1
                            key.cancel();
                        }else {
                            split(buffer);
                            if (buffer.position() == buffer.limit()){
                                // buffer.capacity() 返回缓冲区的容量
                                ByteBuffer newBuffer = ByteBuffer.allocate(buffer.capacity() * 2);
                                buffer.flip();
                                newBuffer.put(buffer);
                                key.attach(newBuffer);
                            }
                        }
                    }catch (IOException e){
                        e.printStackTrace();
                        key.cancel();
                    }
                }
                // 删除已经处理完成的事件
                selectionKeys.remove();
            }
        }
    }

    private static void split(ByteBuffer buffer){
        buffer.flip(); // 切换读模式
        for (int i = 0 ; i < buffer.limit() ; i++ ){
            if ( buffer.get(i) == '\n' ){
                int len = i + 1 - buffer.position();
                ByteBuffer target = ByteBuffer.allocate(len);
                for (int j = 0 ; j < len; j++){
                    target.put(buffer.get());
                }
                ByteBufferUtil.debugAll(target);
            }
        }
        buffer.compact();// 清除已读内容
    }

    /**
     * 用来测试
     */
    @Test
    void ClientTest() throws IOException {
        SocketChannel socketChannel = SocketChannel.open();
        socketChannel.connect(new InetSocketAddress("localhost", 8080));
        socketChannel.write(Charset.defaultCharset().encode("0123456789abcdef3333\n0123456789abcdef3333\n"));
        socketChannel.close();
    }


    /**
     * 用来测试
     */
    @Test
    void WriteServer() throws IOException {
        ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
        serverSocketChannel.configureBlocking(false);
        Selector selector = Selector.open();
        serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
        serverSocketChannel.bind(new InetSocketAddress(8080));
        while (true){
            selector.select();
            Iterator<SelectionKey> selectionKeyIterator = selector.selectedKeys().iterator();
            while ( selectionKeyIterator.hasNext() ){
                SelectionKey selectionKey = selectionKeyIterator.next();
                selectionKeyIterator.remove();
                if ( selectionKey.isAcceptable() ){
                    SocketChannel socketChannel = serverSocketChannel.accept();
                    socketChannel.configureBlocking(false);
                    SelectionKey selectionKey1 = socketChannel.register(selector, SelectionKey.OP_READ);
                    // 向客户端发送大量数据
                    StringBuilder str = new StringBuilder();
                    for (int i = 0; i < 30000000; i++) {
                        str.append("a");
                    }
                    // 将字符串转换为 bytebuffer
                    ByteBuffer buffer = Charset.defaultCharset().encode(str.toString());
                    int len = socketChannel.write(buffer);
                    log.debug("已经写入了{}字节",len);
                    if ( buffer.hasRemaining() ){
                        // 在原有关注事件的基础上，多关注 写事件
                        selectionKey1.interestOps(selectionKey1.interestOps() + SelectionKey.OP_WRITE);
                        // 把 buffer 作为附件加入 selectionKey1
                        selectionKey1.attach(buffer);
                    }
                }else if ( selectionKey.isWritable() ){
                    ByteBuffer buffer = (ByteBuffer) selectionKey.attachment();
                    SocketChannel sc = (SocketChannel) selectionKey.channel();
                    int write = sc.write(buffer);
                    log.debug("接续写入了{}字节",write);
                    if (!buffer.hasRemaining()) { // 写完了
                        selectionKey.interestOps(selectionKey.interestOps() - SelectionKey.OP_WRITE);
                        selectionKey.attach(null);
                    }
                }
            }
        }
    }

    /**
     * 用来测试
     */
    @Test
    void WriteClient() throws IOException {
        SocketChannel socketChannel = SocketChannel.open();
        socketChannel.connect(new InetSocketAddress("localhost", 8080));
        // 接收数据
        int count = 0;
        while (true) {
            ByteBuffer buffer = ByteBuffer.allocate(1024 * 1024);
            count += socketChannel.read(buffer);
            buffer.clear();
            log.debug("已经读入入了{}字节",count);
        }
    }

    /**
     * 用来测试
     */
    @Test
    void BossServer() throws IOException {
        Thread.currentThread().setName("Boss");
        ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
        serverSocketChannel.configureBlocking(false);
        Selector selector = Selector.open();
        serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
        serverSocketChannel.bind(new InetSocketAddress(8080));
        // Work
        Work[] works = new Work[2];
        for (int i = 0; i < works.length; i++) {
            works[i] = new Work("Work" + i);
        }
        // 计数器
        AtomicInteger index = new AtomicInteger();
        while (true){
            selector.select();
            Iterator<SelectionKey> selectionKeyIterator = selector.selectedKeys().iterator();
            while ( selectionKeyIterator.hasNext() ){
                SelectionKey selectionKey = selectionKeyIterator.next();
                selectionKeyIterator.remove();
                if ( selectionKey.isAcceptable() ){
                    SocketChannel socketChannel = serverSocketChannel.accept();
                    socketChannel.configureBlocking(false);
                    works[index.getAndIncrement() % works.length].register(socketChannel);
                }
            }
        }
    }

    static class Work implements Runnable{

        private Thread thread;

        private Selector selector;

        private String name;

        private ConcurrentLinkedQueue<Runnable> queue = new ConcurrentLinkedQueue<>();

        // 是否已经被初始化
        private volatile boolean start = false;

        public Work(String name) {
            this.name = name;
        }

        // 初始化 Thread 和 Selector
        public void register(SocketChannel socketChannel) throws IOException {
            if ( !start ){
                thread = new Thread(this, name);
                thread.start();
                selector = Selector.open();
                start = true;
            }
            // 向队列添加了任务，但是并未立即执行
            queue.add(() -> {
                try {
                    socketChannel.register(selector, SelectionKey.OP_READ, null);
                }catch (IOException e) {
                    e.printStackTrace();
                }
            });
            selector.wakeup(); // 唤醒下边的select
        }

        @Override
        public void run() {
            while (true) {
                try {
                    selector.select();
                    Runnable task = queue.poll();
                    if ( task != null ){
                        task.run();
                    }
                    Iterator<SelectionKey> selectionKeyIterator = selector.selectedKeys().iterator();
                    while ( selectionKeyIterator.hasNext() ){
                        SelectionKey key = selectionKeyIterator.next();
                        selectionKeyIterator.remove();
                        if ( key.isReadable() ){
                            try{
                                ByteBuffer buffer = ByteBuffer.allocate(16);
                                SocketChannel socketChannel = (SocketChannel) key.channel();
                                long len = socketChannel.read(buffer);
                                if ( len == -1 ){
                                    key.cancel();
                                }else {
                                    buffer.flip();
                                    log.debug("线程名{}",name);
                                    ByteBufferUtil.debugAll(buffer);
                                }
                            }catch (IOException e){
                                e.printStackTrace();
                                key.cancel();
                            }
                        }
                    }
                }catch (IOException e){
                    e.printStackTrace();
                }
            }
        }

    }

    /**
     * 用来测试
     */
    @Test
    void BossClientTest() throws IOException {
        SocketChannel socketChannel = SocketChannel.open();
        socketChannel.connect(new InetSocketAddress("localhost", 8080));
        socketChannel.write(Charset.defaultCharset().encode("0123456"));
        socketChannel.close();
    }

    /**
     * 用来测试
     */
    @Test
    void UDPServer() throws IOException {
        try (DatagramChannel channel = DatagramChannel.open()) {
            channel.socket().bind(new InetSocketAddress(8080));
            System.out.println("waiting...");
            ByteBuffer buffer = ByteBuffer.allocate(16);
            channel.receive(buffer);
            buffer.flip();
            ByteBufferUtil.debugAll(buffer);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 用来测试
     */
    @Test
    void UDPClient() throws IOException {
        try (DatagramChannel channel = DatagramChannel.open()) {
            InetSocketAddress address = new InetSocketAddress("localhost", 8080);
            ByteBuffer buffer = StandardCharsets.UTF_8.encode("hello");
            channel.send(buffer, address);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

}
